package com.atguigu.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV1;
import com.atguigu.realtime.bean.TableProcess;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.FlinkSinkUtil;
import com.atguigu.realtime.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

public class Dwd_BaseDBApp extends BaseAppV1 {
    public static void main(String[] args) {
        new Dwd_BaseDBApp().init(
                3002,
                2,
                "Dwd_BaseDBApp",
                Constant.TOPIC_ODS_DB
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        SingleOutputStreamOperator<JSONObject> etledStream = etl(stream);
        SingleOutputStreamOperator<TableProcess> tpStream = readTableProcess(env);
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream = connect(etledStream, tpStream);
        resultStream = delNoNeedColumns(resultStream);
        writeToKafka(resultStream);

    }

    private void writeToKafka(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getKafkaSink());
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> delNoNeedColumns(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream) {
        return resultStream.map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
            @Override
            public Tuple2<JSONObject, TableProcess> map(Tuple2<JSONObject, TableProcess> t) throws Exception {
                List<String> list = Arrays.asList(t.f1.getSinkColumns().split(","));

                t.f0.keySet().removeIf(c -> !list.contains(c));
                return t;
            }
        });
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(SingleOutputStreamOperator<JSONObject> dataStream, SingleOutputStreamOperator<TableProcess> tpStream) {
        // 1. 先把 tpStream 转成广播流
        // key:
        MapStateDescriptor<String, TableProcess> tpStateDesc = new MapStateDescriptor<>("tpState ", String.class, TableProcess.class);
        BroadcastStream<TableProcess> tpBcStream = tpStream.broadcast(tpStateDesc);
        // 2. 数据流去 connect 广播流
        return dataStream
                .connect(tpBcStream)
                .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>() {

                    HashMap<String, TableProcess> tpMap;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        Connection conn = JdbcUtil.getMysqlConnection();
                        List<TableProcess> tpList = JdbcUtil.queryList(conn,
                                "select * from edu_config.table_process",
                                null,
                                TableProcess.class,
                                true);

                        tpMap = new HashMap<>();
                        for (TableProcess tp : tpList) {
                            if ("dwd".equals(tp.getSinkType())) {
                                String key = tp.getSourceTable() + ":" + tp.getSinkType() + ":" + (tp.getSinkExtend() == null ? "" : tp.getSinkExtend());
                                tpMap.put(key, tp);
                            }
                        }
                    }

                    @Override
                    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        //TODO 根据来的数据, 找到对应的配置信息
                        String table = value.getString("table");
                        String type = value.getString("type");
                        String key = table + ":" + type + ":";

                        ReadOnlyBroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                        TableProcess tp = state.get(key);
                        if (tp == null) {
                            //再去map中看看有没有
                            tp = tpMap.get(key);
                        }

                        if (tp != null) {
//                            value.getJSONObject("data").put("ts", value.getLong("ts"));
                            out.collect(Tuple2.of(value.getJSONObject("data"), tp));
                        }
                    }

                    @Override
                    public void processBroadcastElement(TableProcess tp, Context ctx, Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        //TODO 处理广播数据: 把配置信息写入到广播状态中
                        BroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                        String key = tp.getSourceTable() + ":" + tp.getSourceType() + ":" + (tp.getSinkExtend() == null ? "" : tp.getSinkExtend());
                        state.put(key, tp);

                        if ("d".equals(tp.getOp())) {
                            state.remove(key);
                        }
                    }
                });
    }

    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment env) {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop162")
                .port(3306)
                .databaseList("edu_config")
                .tableList("edu_config.table_process")
                .username("root")
                .password("aaaaaa")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        return env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-source")
                .map(new MapFunction<String, TableProcess>() {
                    @Override
                    public TableProcess map(String value) throws Exception {
                        JSONObject obj = JSON.parseObject(value);
                        String op = obj.getString("op");
                        TableProcess tp = null;
                        if ("d".equals(op)) {
                            // 如果是删除, 取 before, 封装到 TableProcess 中
                            tp = obj.getObject("before", TableProcess.class);
                        } else {
                            tp = obj.getObject("after", TableProcess.class);
                        }
                        tp.setOp(op);
                        return tp;
                    }
                })
                .filter( tp -> "dwd".equals(tp.getSinkType())); // 过滤出 dwd 的配置
    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {
                        try {
                            JSONObject obj = JSON.parseObject(value.replaceAll("bootstrap-", ""));
                            String type = obj.getString("type");
                            String data = obj.getString("data");
                            // 证明 json 格式是对的
                            return "edu".equals(obj.getString("database"))
                                    && ("insert".equals(type) || "update".equals(type))
                                    && data != null
                                    && data.length() > 2;

                        } catch (Exception e) {
                            System.out.println("你的 json 格式数据异常: " + value);
                            return false;
                        }
                    }
                })
                .map(json -> JSON.parseObject(json.replaceAll("bootstrap-", "")));
    }
}
